#include "vl_thread_pool.h"

using namespace VL;

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 15:18:15
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * VLFuture
 * 
 */


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:17:36
 * 
 * 中断当前 Future 任务
 */
void VLFuture::interrupt() {
	std::lock_guard<std::mutex> locker(this->mutex);
	if(this->completed || this->interrupted) {
		return;
	}
	this->interrupted = true;
	this->cond.notify_all();
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 08:17:08
 * 
 * 等待结果
 */
void VLFuture::wait() {
	std::unique_lock<std::mutex> locker(this->mutex);
	if(this->completed || this->interrupted) {
		return;
	}
	this->cond.wait(locker);
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 08:17:50
 * 
 * 判断 Future 是否已经完成
 * 
 * @return 如果成功返回 true
 */
bool VLFuture::isCompleted() {
	return this->completed;
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 08:17:50
 * 
 * 判断 Future 是否已经中断
 * 
 * @return 如果中断返回 true
 */
bool VLFuture::isInterrupted() {
	return this->interrupted;
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:28:12
 * 
 * 获取数据, 否则抛出异常
 */
void VLFuture::getOrThrow() {
	return;
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 08:16:38
 * 
 * 保护构造函数
 */
VLFuture::VLFuture(): completed(false), interrupted(false) {

}

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 15:26:19
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * VLThreadRunnable
 * 
 */

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:11:24
 * 
 * 构造方法
 * 
 * @param runnable 可执行回调
 */
VLThreadRunnable::VLThreadRunnable(Runnable runnable): buffer(nullptr), futurePtr(nullptr), runnable(runnable), futureRunnable(nullptr) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:11:46
 * 
 * 构造方法
 * 
 * @param runnable 携带 Future 的可执行回调
 */
VLThreadRunnable::VLThreadRunnable(FutureRunnable runnable): buffer(nullptr), futurePtr(nullptr), runnable(nullptr), futureRunnable(runnable) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:09:54
 * 
 * 线程执行方法
 */
void VLThreadRunnable::threadRun() {
	if(runnable) {
		runnable(this->buffer);
	} else if(futureRunnable) {
		futureRunnable(this->futurePtr, this->buffer);
	}
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:41:45
 * 
 * 中断
 */
void VLThreadRunnable::interrupt() {
	if(this->futurePtr.get() != nullptr) {
		this->futurePtr->interrupt();
	}
}

/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 16:29:04
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * ThreadFutureManager
 * 
 */

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:25:55
 * 
 * 构造函数
 */
ThreadFutureManager::ThreadFutureManager() {

}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:27:44
 * 
 * 获取数据, 否则抛出异常
 */
void ThreadFutureManager::getOrThrow() {
	for(auto iter = this->futureList.begin() ; iter != this->futureList.end() ; iter ++) {
		(*iter)->getOrThrow();
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:26:58
 * 
 * 添加 Future
 * 
 * @param futurePtr Future 指针
 */
void ThreadFutureManager::addFuture(const std::shared_ptr<VLFuture>& futurePtr) {
	this->futureList.push_back(futurePtr);
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 16:27:09
 * 
 * 获取 Future
 * 
 * @param index Future 下标
 * 
 * @return Future 指针
 */
std::shared_ptr<VLFuture> ThreadFutureManager::getFuture(unsigned int index) {
	if(index >= this->futureList.size()) {
		return std::shared_ptr<VLFuture>(nullptr);
	}

	return this->futureList.at(index);
}


/**
 * +======================================================================================================================+
 * @Author: CimZzz
 * @Date: 2022-08-21 15:28:31
 * 
 * 代码分割块，以下内容涉及: 
 * 
 * VLThreadPool
 * 
 */


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 01:08:19
 * 
 * 线程池单例对象
 */
VLThreadPool VLThreadPool::instance;

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:10:39
 * 
 * 初始化线程池
 * 
 * @param threadCount 线程数
 */
void VLThreadPool::initThreadPool(unsigned int threadCount) {
	if(this->isInit) {
		return;
	}
	this->isInit = true;
	// 构建线程池
	for(unsigned int i = 0 ; i < threadCount ; i ++) {
		this->threadList.push_back(std::shared_ptr<std::thread>(new std::thread([&](){
			// 线程循环
			std::shared_ptr<Buffer> buffer(new Buffer);
			while(this->isRunning) {
				std::shared_ptr<VLThreadRunnable> runnable(nullptr);
				{
					std::unique_lock<std::mutex> locker(this->mutex);
					if(!this->isRunning) {
						return;
					}

					if(this->runnableQueue.empty()) {
						// 队列中不存在回调, 等待
						this->cond.wait(locker);
					}

					if(!this->runnableQueue.empty()) {
						// 队列中不存在回调, 等待
						runnable = this->runnableQueue.front();
						this->runnableQueue.pop();
					}
				}

				if(runnable.get() != nullptr) {
					buffer->reset();
					runnable->buffer = buffer.get();
					runnable->threadRun();
					runnable.reset();
					std::lock_guard<std::mutex> locker(this->mutex);
					this->taskCount --;
					if(this->taskCount == 0) {
						// 全部任务完成
						this->taskDoneCond.notify_all();
					}
				}
			}
		})));
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:12:41
 * 
 * 执行线程回调
 * 
 * @param runnable 线程回调
 */
void VLThreadPool::exec(VLThreadRunnable::Runnable runnable) {
	auto ptr = std::shared_ptr<VLThreadRunnable>(new VLThreadRunnable(runnable));
	this->exec(ptr);
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:58:00
 * 
 * 等待全部任务完成
 */
void VLThreadPool::waitForAllTaskDone() {
	if(!this->isRunning) {
		return;
	}
	std::unique_lock<std::mutex> locker(this->mutex);
	if(this->taskCount == 0) {
		return;
	}
	this->taskDoneCond.wait(locker);
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 17:07:33
 * 
 * 将后续任务添加到等待队列中
 */
void VLThreadPool::pendingTask() {
	this->isPending = true;
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 17:07:51
 * 
 * 将等待队列中的任务放置到主消息队列中
 */
void VLThreadPool::continueTask() {
	if(this->isPending) {
		this->isPending = false;
		std::lock_guard<std::mutex> locker(this->mutex);
		while(!this->pendingQueue.empty()) {
			auto runnable = this->pendingQueue.front();
			this->pendingQueue.pop();
			this->runnableQueue.push(runnable);
			this->taskCount ++;
		}
		this->cond.notify_all();
	}
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:14:57
 * 
 * 中断线程池中其他任务
 */
void VLThreadPool::clearTask() {
	std::lock_guard<std::mutex> locker(this->mutex);
	while(!this->runnableQueue.empty()) {
		auto runnable = this->runnableQueue.front();
		runnable->interrupt();
		this->runnableQueue.pop();
		this->taskCount --;
	}
	while(!this->pendingQueue.empty()) {
		auto runnable = this->pendingQueue.front();
		runnable->interrupt();
		this->pendingQueue.pop();
	}
	if(this->taskCount == 0) {
		this->taskDoneCond.notify_all();
	}
}


/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:29:34
 * 
 * 获取线程池单例指针
 * 
 * @return 线程池指针
 */
VLThreadPool* VLThreadPool::getInstance() {
	return &VLThreadPool::instance;
}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 01:07:12
 * 
 * 构造函数
 */
VLThreadPool::VLThreadPool(): isInit(false), isRunning(true), taskCount(0), isPending(false) {

}

/**
 * @Author: CimZzz
 * @Date: 2022-08-21 01:08:04
 * 
 * 析构函数
 */
VLThreadPool::~VLThreadPool() {
	this->isRunning = false;
	this->clearTask();
	this->cond.notify_all();
	for(auto threadIter = this->threadList.begin() ; threadIter != this->threadList.end() ; threadIter ++) {
		auto threadPtr = (*threadIter);
		if(threadPtr->joinable()) {
			threadPtr->join();
		}
	}
}



/**
 * @Author: CimZzz
 * @Date: 2022-08-21 15:40:01
 * 
 * 执行线程回调
 * 
 * @param runnable 线程回调
 */
void VLThreadPool::exec(const std::shared_ptr<VLThreadRunnable>& runnable) {
	if(!this->isRunning) {
		runnable->interrupt();
		return;
	}

	if(this->isPending) {
		this->mutex.lock();
		this->pendingQueue.push(runnable);
		this->mutex.unlock();
	} else {
		this->mutex.lock();
		this->taskCount ++;
		this->runnableQueue.push(runnable);
		this->mutex.unlock();
		this->cond.notify_all();
	}
}